package chapter2;

import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.Consumer;
import io.reactivex.observables.ConnectableObservable;
import io.reactivex.schedulers.Schedulers;

import java.util.concurrent.TimeUnit;

/**
 * @author shaopeng.wei
 * @since 2019-01-14 14:45
 * Purpose
 */
public class ConnectObservable2ColdObservable {
    public static void main(String[] args) throws InterruptedException {
        Consumer<Long> subscriber1 = new Consumer<Long>() {
            @Override
            public void accept(Long aLong) throws Exception {
                System.out.println("subscriber1:" + aLong);
            }
        };

        Consumer<Long> subscriber2 = new Consumer<Long>() {
            @Override
            public void accept(Long aLong) throws Exception {
                System.out.println("subscriber2:" + aLong);
            }
        };


        Observable<Long> longObservable = Observable.create(new ObservableOnSubscribe<Long>() {
            @Override
            public void subscribe(ObservableEmitter<Long> e) throws Exception {
                Observable.interval(5, TimeUnit.MILLISECONDS, Schedulers.io()).take(Integer.MAX_VALUE).subscribe(e::onNext);
            }
        }).observeOn(Schedulers.newThread());
        System.out.println("默认cold Observable，subscriber1先发送10ms，预期subscriber2不受干扰从0开始");
        Disposable subscribe1 = longObservable.subscribe(subscriber1);
        Thread.sleep(10L);
        Disposable subscribe2 = longObservable.subscribe(subscriber2);


        Thread.sleep(20L);
        subscribe1.dispose();
        subscribe2.dispose();

        System.out.println("和预期一直，并没有因为subscriber1先执行而导致subscriber2跟着subscriber1开始");
        System.out.println("————————————————————————————————————————————————————————————————————");
        System.out.println("Cold Observable转化为Hot 不会重新开始，所有subscriber2会跟着subscriber1的数，不会从0开始。");
        ConnectableObservable<Long> publish = longObservable.publish();
        publish.connect();

        Disposable disposable1 = publish.subscribe(subscriber1);
        Thread.sleep(10L);
        Disposable disposable2 = publish.subscribe(subscriber2);

        Thread.sleep(20L);
        disposable1.dispose();
        disposable2.dispose();
        System.out.println("发现和预期一致");
        System.out.println("————————————————————————————————————————————————————————————————————");
        System.out.println("预期和cold Observable一样");
        Observable<Long> observable = publish.refCount();
        disposable1 = observable.subscribe(subscriber1);
        Thread.sleep(20L);
        disposable2 = observable.subscribe(subscriber2);

        Thread.sleep(20L);
        disposable1.dispose();
        disposable2.dispose();
        System.out.println("发现预期并不一样。因为如果不是所有的订阅者/观察者取消了订阅，而只是部分取消，则部分的订阅者/观察者重新开始订阅时，不会从头开始数据流");

    }
}
